package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import redis.clients.jedis.JedisCluster;

import java.io.Serializable;
import java.util.Map;

/**
 * @author : GW00280745
 * @date : 2022/8/15 8:58
 * @description : 移动平均值计算方式 : ( map_table_val + val1 + val2 ) / 3
 * mapVal  map表中得数据格式:{move_avg:val,tid1:temp_acc_val,tid2:temp_acc_val}
 */
public class MapTempAvg implements Serializable {

    private String redisNameSpaceTemp = ChargePredictFields.ins().redisNameSpaceTemp;
    public MapTempAvg(String redisNameSpaceTemp){

    }
    public void process(JSONObject line,
                        MapState<String, Map> latestMoveAvgState,
                        KeyedBroadcastProcessFunction.ReadOnlyContext ctx,
                        JedisCluster jedis){
        try {
            avg(line,latestMoveAvgState,ctx,jedis);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    private MapStateDescriptor<String, Map<String, Object>> confInfo = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));

    private void avg(JSONObject line,
                     MapState<String, Map> latestMoveAvgState,
                     KeyedBroadcastProcessFunction.ReadOnlyContext ctx,
                     JedisCluster jedis) throws Exception{
        ReadOnlyBroadcastState<String, Map<String, Object>> broadcastState = ctx.getBroadcastState(confInfo);
        Map<String, Object> confMap = broadcastState.get(null);
        //1.获取当前记录得温升速率
        String currentRecordAcc = line.getString(ChargePredictFields.ins().tempAccField);
        if (StringUtils.isBlank(currentRecordAcc)){
//            line.put(StdFieldsConstant.stdField_moveAvgPoint,latestMoveAvg)
//            line.put(StdFieldsConstant.stdField_tempAccTempRange,latestMoveAvgState.get(StdFieldsConstant.stdField_tempAccTempRange))
//            line.put(StdFieldsConstant.stdField_tempAccBatterRange,latestMoveAvgState.get(StdFieldsConstant.stdField_tempAccBatterRange))
            return;
        }
        //2.获取map表中得移动平均值，需要和redis交互 10.255.128.239:6370
        String vehicle = AxisInfo.singleInstance().axisTempAcc(line,null,null,confMap,true);
//        String vehicle = ChargeUtil.single().axisTempAcc(line,null,null)
        String axisDefault = ChargeInfo.single().axisTempAccDefault(line, null, null);
        String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
        String redisKey = redisNameSpaceTemp.concat(vehicle);
        String redisKeyDefault = redisNameSpaceTemp.concat(axisDefault);
        String mapValJson = ChargeInfo.single().getRedisHval(jedis,redisKey,vinVal);
        String mapValJsonAvg = ChargeInfo.single().getRedisHval(jedis,redisKeyDefault,"avg");
        JSONObject mapVal = StringUtils.isBlank(mapValJson)?new JSONObject():JSONObject.parseObject(mapValJson);
        JSONObject mapValAvg = StringUtils.isBlank(mapValJsonAvg)?new JSONObject():JSONObject.parseObject(mapValJsonAvg);
        //3.计算最新得移动平均值
        String moveAvg = moveAvg(mapVal,currentRecordAcc,line);
        String moveAvgAvg = moveAvg(mapValAvg,currentRecordAcc,new JSONObject());
        //4.更新到redis
        jedis.hset(redisKey,vinVal,moveAvg);
        jedis.hset(redisKeyDefault,"avg",moveAvgAvg);
        System.out.println(String.format("tempacc #&# statebackend #&# update #&# hashKey=%s #&# fieldKey=%s #&# query value=%s #&# update value=%s",vehicle,vinVal,mapValJson,moveAvg));
        //4.1 更新到state中
        avgVal2State(latestMoveAvgState, vinVal, redisKey, moveAvg);
        avgVal2State(latestMoveAvgState, "avg", redisKeyDefault, moveAvgAvg);
        //5.更新数据状态
        JSONObject moveAvgJo = JSONObject.parseObject(moveAvg);
        String tempMoveAvg = moveAvgJo.getString("tempMoveAvg");
        line.put(ChargePredictFields.ins().stdFieldMoveAvgPoint,tempMoveAvg);
        //移动平均值计算过程
        moveAvgJo.remove("tempMoveAvg");
        line.put(ChargePredictFields.ins().stdFieldMoveAvgProcess,moveAvgJo.toJSONString());
        //end
    }

    private void avgVal2State(MapState<String, Map> latestMoveAvgState, String vinVal, String redisKey, String moveAvg) throws Exception {
        Map map = latestMoveAvgState.get(redisKey);
        JSONObject jsonObj =(JSONObject) map.getOrDefault(redisKey, new JSONObject());
        jsonObj.put(vinVal, moveAvg);
        latestMoveAvgState.put(redisKey,jsonObj);
    }

    /**
     * @description: 移动平均值计算方式 : ( map_table_val + val1 + val2 ) / 3
     * @author:  GW00280745
     * @date: 2022/5/9 10:46
     * @param: mapVal  map表中得数据格式:{move_avg:val,tid1:temp_acc_val,tid2:temp_acc_val}
     * @param: val vin#tid#acc
     * @return: java.lang.String
     **/
    private String moveAvg(JSONObject mapVal,String val,JSONObject dwsLine){
        //将更新得mapVal转成字符串返回
        String startsWith = ChargePredictFields.ins().tempMoveAvgField;
        String pointField = ChargePredictFields.ins().stdFieldMoveAvgPoint;
        String errField=ChargePredictFields.ins().stdFieldTempAccErr;
        String avgField = ChargePredictFields.ins().tempMoveAvgField;
        String moveAvg = ChargeInfo.single().moveAvg(mapVal, val, dwsLine, startsWith, pointField, errField, avgField);
        return moveAvg;
    }
}
